package plgmqtt

import (
	"dgo/MyProj/PrjHttpUtils"
	"dgo/goutils/utils"
	"dgo/plg/plgnet"
	"gitee.com/ymofen/supervalue"
	"time"
)

/*
*

	会分发所有收到的数据给订阅者
*/
type PlgMqttSub struct {
	PlgBase
	msglst       *utils.SyncCycleList
	mqtt         *MqttInner
	sublst       utils.GSyncMapGroup
	monitor_recv *utils.MonitorSpeedSizeRec
}

/*
*

	  {
	    "id": "xxx",
	    "type": "plg.mqtt.sub",
	    "conf": {
	        "topics":"多个订阅用,分割",   // 如果存在不能进行主动订阅，或者会被覆盖
			"clientid":"",
			"addr":"tcp://139.199.7.76:52883",
			"user":"",
			"pass":"",
	        "output":{}
	    }
	  }
*/
func (this *PlgMqttSub) ReloadConf(conf *wrapper.SuperValue) error {
	if conf == nil {
		return nil
	}

	if this.msglst == nil {
		this.msglst = utils.NewSyncCycleList()
	}

	if this.monitor_recv == nil {
		this.monitor_recv = utils.NewMonitorSpeedSizeRec()
		utils.Monitor.AddMonitor(this.monitor_recv)
	}

	if this.mqtt == nil {
		this.mqtt = NewMqttInner()
	}
	this.mqtt.ReloadConf(conf)
	conf.CheckExistsStrFunc("topics", func(strv string) {
		this.mqtt.Resubscribe(strv)
	})

	this.ReloadOutputConf(conf.ValueByName("output"))

	this.mqtt.OnMessage = func(sender *MqttInner, channelid string, data []byte) {
		this.monitor_recv.DeltaBuf(data)
		var isBreak bool = false
		this.sublst.Range(func(key, value interface{}) bool {
			if cb, ok := value.(utils.SubscribeCallback); ok {
				cb(this, channelid, key, data, nil, &isBreak)
			}
			return !isBreak
		})
		this.OutputEx.DispatchData(channelid, this, data, nil)
	}

	utils.DefaultPatrolTask().AddTask(this, time.Second*30, func(id interface{}, args ...interface{}) {
		//utils.DefaultWorkers().PostTaskFunc(this, this.innerWork)
	})
	return nil
}

func (this *PlgMqttSub) Close() (err error) {
	this.mqtt.Close()
	utils.Monitor.RemoveMonitorRec(this.monitor_recv)
	this.PlgBase.Close()
	utils.DefaultPatrolTask().DelTask(this)
	return
}

func (this *PlgMqttSub) OnHttpStatus(rec *plgnet.PlgHttpRec) (rval interface{}, errcode int, err error) {
	itype := rec.QryAsInt("type", 0)
	if itype == 99 {
		json := wrapper.NewSVObject()
		json.SetKeyValue("recv", this.monitor_recv.Info())
		json.SetKeyValue("sub-topics", this.mqtt.SubTopics(" "))
		json.SetKeyValue("mqtt-status", this.mqtt.StatusString())
		json.SetKeyValues("now", utils.NowString())
		if rec.QryAsInt("json", 0) == 1 {
			PrjHttpUtils.ResponseAsJSON(rec.RespW, json.AsJSONStringEx(false, true))
		} else {
			PrjHttpUtils.JSON2TableRec(rec.RespW, "", json.AsJSONString(false))
		}
	} else if itype == 98 {
		rec.ResponseSyncListNEx(this.msglst)
	} else if itype == 1 {
		json := wrapper.NewSVArray()
		//searchval := rec.QryAsString("searchval", "")
		//sn := 0
		//maxn := rec.QryAsInt("n", 100)
		//this.lst.ListRangeV2(func(key, val interface{}) bool {
		//	pojo := val.(*VrsSitePojo)
		//	ok := true
		//	if len(searchval) > 0 {
		//		ok = strings.Contains(pojo.id, searchval)
		//	}
		//	return ok
		//}, func(k0, v0, k1, v1 interface{}) bool {
		//	p0 := v0.(*VrsSitePojo)
		//	p1 := v1.(*VrsSitePojo)
		//	return p0.id < p1.id
		//}, func(key, val interface{}) bool {
		//	sn++
		//	pojo := val.(*VrsSitePojo)
		//	itm := json.AppendObject()
		//	itm.SetKeyValue("sn", sn)
		//	itm.SetKeyValue("id", pojo.id)
		//	return sn < maxn
		//})

		if rec.QryAsInt("json", 0) == 1 {
			PrjHttpUtils.ResponseAsJSON(rec.RespW, json.AsJSONStringEx(false, false))
		} else {
			PrjHttpUtils.JSON2Table(rec.RespW, "", json.AsJSONString(false))
		}
	} else {
		var sb utils.BytesBuilder
		PrjHttpUtils.AppendIndexBody(&sb, "status", "status?type=99")
		PrjHttpUtils.AppendIndexBody(&sb, "msglist", "status?type=98&n=100&searchval=")
		PrjHttpUtils.AppendIndexBody(&sb, "lst", "status?type=1&n=100&searchval=")
		PrjHttpUtils.ResponseBodyHtml(rec.RespW, sb.String())
	}
	return
}

/*
*

	忽略channelid
*/
func (this *PlgMqttSub) SubscribeChannel(channelid string, id interface{}, cb utils.SubscribeCallback) {
	this.sublst.Set(id, cb)
}

func (this *PlgMqttSub) UnsubscribeChannel(channelid string, id interface{}) bool {
	this.sublst.Remove(id)
	return true
}

func (this *PlgMqttSub) LogInfoMessage(s string, args ...interface{}) {
	this.msglst.LogMsg(s, args...)
	utils.Log.Msgf(this.PlugId, utils.LogLV_INFO, 1, s, args...)
}

func (this *PlgMqttSub) LogWarnMessage(s string, args ...interface{}) {
	this.msglst.LogMsg(s, args...)
	utils.Log.Msgf(this.PlugId, utils.LogLV_WARN, 1, s, args...)
}

func CreatePlgMqttSubFunc(args ...interface{}) (rval interface{}, err error) {
	obj := &PlgMqttSub{}

	if len(args) > 0 {
		if str, ok := args[0].(string); ok {
			obj.PlugId = str
		}
	}
	rval = obj
	return
}
